package org.com.lcuiot.config;

import org.com.lcuiot.constant.LcuIotConst;
import org.com.lcuiot.service.AcceptDataService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @author gaoge
 * @since 2022/10/27 14:50
 */
@Component
public class IotMsgListener implements StreamListener<String, MapRecord<String, String, String>> {
    private static final Logger logger = LoggerFactory.getLogger(IotMsgListener.class);

    private RedisStream redisStream;

    private AcceptDataService acceptDataService;

    @Autowired
    public IotMsgListener(RedisStream redisStream, AcceptDataService acceptDataService) {
        this.redisStream = redisStream;
        this.acceptDataService =  acceptDataService;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        // 接收到消息
        logger.info("----------message {}", message.getId());
        logger.info("---------stream {}", message.getStream());
        logger.info("------------body {}",  message.getValue());
        Map<String, String> val = message.getValue();

        // 处理数据并入库
        acceptDataService.save2DB(val);

        // 确认消息
        redisStream.ack(LcuIotConst.IOT_STREAM, LcuIotConst.IOT_STREAM_GROUP, message.getId());

        // trim stream的长度
        if(redisStream.len(LcuIotConst.IOT_STREAM) > LcuIotConst.IOT_STREAM_TRIM_LEN) {
            redisStream.trim(LcuIotConst.IOT_STREAM, LcuIotConst.IOT_STREAM_TRIM_LEN);
        }
    }

}
